package hasthi.container;

import static hasthi.common.constants.HasthiConstants.FAULTCODE_COORDINATOR_MSG_RECIVEDBY_NONCOORDINATOR;
import static hasthi.common.constants.HasthiConstants.FAULT_DETAIL_CURRENT_COORDINATOR;
import static hasthi.common.constants.HasthiConstants.FAULT_DETAIL_CURRENT_COORDINATOR_VAL_RETRY;
import static hasthi.common.constants.HasthiConstants.GET_COORDINATOR_ACTION;
import static hasthi.common.constants.HasthiConstants.LOGGER_NAME;
import static hasthi.common.constants.HasthiConstants.SELF_ADDRESS;
import hasthi.common.BaseContext;
import hasthi.common.ComponetFactory;
import hasthi.common.DistributedLogger;
import hasthi.common.HasthiException;
import hasthi.common.HasthiException.FaultCode;
import hasthi.container.p2p.MulticastContianier;
import hasthi.manager.GetCoordinatorDocument;
import hasthi.manager.GetCoordinatorResponseDocument;
import hasthi.manager.ManagerAddress;
import hasthi.manager.ManagerContext;
import hasthi.manager.ManagerMessageProcessor;
import hasthi.manager.GetCoordinatorResponseDocument.GetCoordinatorResponse;
import hasthi.tools.ManagementRingTools;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.Properties;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * This class represents a Management Node, it finds a free port on the given
 * host and start itself there. TODO right now bootstrap node is hardcoded
 * 
 * @author Srinath Perera(hperera@cs.indiana.edu)
 */

public class ManagmentNode {
    private static Log log = LogFactory.getLog(LOGGER_NAME);

    private ManagerContext managerContext;
    private MessageProcessor managerInstance;
    private DistributedLogger dlogger;

    public ManagmentNode() throws HasthiException {
        this((Properties)null);
    }
    
    public ManagmentNode(Properties properties) throws HasthiException {
        this(new BaseContext(properties));
    }
    
    public ManagmentNode(BaseContext baseContext) throws HasthiException {
        managerContext = new ManagerContext(baseContext);
        managerInstance = createMessageProcessor();
        dlogger = managerContext.getBaseContext().getDlogger();
    }

    public void startManagmentNode() throws HasthiException {
        MulticastContianier p2pContainer = null;
        UnicastContianier soapContainer = null;
        try {
            BaseContext baseContext = managerContext.getBaseContext();
            // Initialize the Containiers and add them to Manager Context
            p2pContainer = ComponetFactory.createMulticastContianier(baseContext, managerInstance);
            p2pContainer.startContainer();
            managerContext.setMulticastContianer(p2pContainer);
            soapContainer = ComponetFactory.createSoapContianier(baseContext,managerInstance);
            soapContainer.startContainer();
            managerContext.setUnicastContainer(soapContainer);

            GetCoordinatorDocument document = GetCoordinatorDocument.Factory.newInstance();
            document.addNewGetCoordinator().setRequesterAddress(managerContext.getMyAddress());

            /**
             * First check am I the first Node to Join. Then Assume I am
             * coordinator. Otherwise keep anycasting till Coordinator is known. If
             * someone retun null, that mean election is going on and We retry after
             * a random wait. This could fail if there was Managers but all died
             * out, but we are ignoring that case for now. If address is null, then
             * a Coordinator is being elected. Wait for random time and retry
             */
            String bootstrapSoapAddress = "http://" + baseContext.getBoostrapHost() + ":"
                    + baseContext.getBoostrapSoapPort();
            MessageContext requestMessage = new SimpleMessageContext(document,bootstrapSoapAddress,
                    GET_COORDINATOR_ACTION);
            GetCoordinatorResponseDocument responseDocument = (GetCoordinatorResponseDocument) soapContainer
                    .sendReceive(requestMessage);
            
            ManagerAddress coordinatorAddress = null;
            if(responseDocument != null){
                coordinatorAddress = responseDocument.getGetCoordinatorResponse().getAddress();
            }

            if (responseDocument != null &&  coordinatorAddress != null 
                    && SELF_ADDRESS.equals(coordinatorAddress.getSoapAddress())) {
                managerContext.becomeCoordinator(System.currentTimeMillis());
                managerContext.setFirstManagerOfRing(true);
                dlogger.info(managerContext.getMyAddress().getSoapAddress()
                        + " assumed Coordinator role by response of bootstrap");
            } else {
                int retryCount = 0;
                String coordinatorSoapAddress = null;
                while(!managerContext.hasCoordinator()){
                    try {
                        /** This is not initialization, I am not first to join the ring * */
                        GetCoordinatorResponse response = ManagementRingTools.findCoordinator(managerContext,
                                -1);
                        /**
                         * When we get here, we know a coordinator. Talk to the coordinator
                         * and Join the Management Ring.
                         */
                        dlogger.debug("receive coordinator " + response.getAddress().getSoapAddress());

                        coordinatorSoapAddress =  response.getAddress().getSoapAddress();
                        ManagementRingTools.joinManagmentRing(managerContext,coordinatorSoapAddress);
                        dlogger.info("Manager "+managerContext +" Started And Joined Coordination Ring");
                        System.out.println("Manager Started And Joined Coordination Ring Sucessfully");
                        return;
                    } catch (Throwable e) {
                        log.error(e);
                        if(e instanceof HasthiException 
                            && ((HasthiException)e).getFaultCode().equals(FaultCode.UnsupportedMessage)){
                            if(e.getMessage().contains(FAULTCODE_COORDINATOR_MSG_RECIVEDBY_NONCOORDINATOR)){
                                String newCoordinatorSoapAddress = ((HasthiException)e).getProperty(FAULT_DETAIL_CURRENT_COORDINATOR);
                                if(newCoordinatorSoapAddress != null){
                                    if(!FAULT_DETAIL_CURRENT_COORDINATOR_VAL_RETRY.equals(coordinatorSoapAddress)
                                            && !newCoordinatorSoapAddress.equals(coordinatorSoapAddress)){
                                        try {
                                            ManagementRingTools.joinManagmentRing(managerContext,coordinatorSoapAddress);
                                        } catch (Throwable e1) {
                                            log.error(e1);
                                        }
                                    }
                                }
                            }
                        }
                        if(!managerContext.hasCoordinator()){
                            if(retryCount >= 5){
                                p2pContainer.stopContainer();
                                soapContainer.stopContainer();
                                managerContext.getManagementDemon().stopDeamon();
                                throw new HasthiException(e,FaultCode.InitalizationError);
                            }else{
                                retryCount++;
                                log.info("retry Join to management ring "+ (retryCount+1) + " time already tried");
                                try {
                                    Thread.sleep(60000);
                                } catch (InterruptedException e1) {
                                    e1.printStackTrace();
                                }
                            }
                        }
                    }
                }
            }
        }catch (Throwable e) {
            log.fatal("Failed to start manager ", e);
            if(p2pContainer != null){
                try{
                    p2pContainer.stopContainer();
                }catch (Exception e1) {}
            }
            if(soapContainer != null){
                try{
                    soapContainer.stopContainer();
                }catch (Exception e1) {}
            }
            managerContext.getManagementDemon().stopDeamon();
            if(e instanceof HasthiException){
                throw (HasthiException)e;    
            }
            throw new HasthiException(e,FaultCode.InitalizationError);
        }
    }

    
    protected MessageProcessor createMessageProcessor() throws HasthiException{
        return new ManagerMessageProcessor(managerContext);
    }
    

    public ManagerContext getManagerContext() {
        return managerContext;
    }
    
    public static void main(String[] args) throws HasthiException {
        try {
            Properties properties = null;
            ManagmentNode node;
            if(args.length > 0){
                properties =  new Properties();
                FileInputStream propertyStream = new FileInputStream(args[0]);
                properties.load(propertyStream);
                propertyStream.close();
                node = new ManagmentNode(properties);
            }else{
                node = new ManagmentNode();
            }
            node.startManagmentNode();
        } catch (FileNotFoundException e) {
            throw new HasthiException(e,FaultCode.LocalError);
        } catch (IOException e) {
            throw new HasthiException(e,FaultCode.LocalError);
        }
    }
}
